package buaa.qel.rpc.client.core;

import buaa.qel.rpc.client.handler.RpcClientHandler;
import buaa.qel.rpc.client.handler.RpcClientChannelInitializer;
import buaa.qel.rpc.client.loadbalance.RpcLoadBalance;
import buaa.qel.rpc.client.loadbalance.impl.RpcLoadBalanceRoundRobin;
import buaa.qel.rpc.common.info.RegistryNodeInfo;
import buaa.qel.rpc.common.info.RpcServiceInfo;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;

import java.net.InetSocketAddress;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.*;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 单例连接管理器
 * 负责提交和保持连接，以及负载均衡
 */
@Slf4j
public class ConnectionManager {
    //默认处理器个数×2
    private final EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
    //用于提交netty连接
    private final ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
            Runtime.getRuntime().availableProcessors(),
            Runtime.getRuntime().availableProcessors()*2,
            300L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(1000));
    //每个host+port（下面有一堆服务）对应一个handler,一个handler处理一个地址的服务
    //一个地址所有rpc的请求发送和结果接收都由这个handler处理
    private final Map<RegistryNodeInfo, RpcClientHandler> connectedServerNodes = new ConcurrentHashMap<>();
    private final Map<String, List<RpcClientHandler>> availableHandlerForService = new ConcurrentHashMap<>();
    //目前已经发现且已经连接的服务器
    private final CopyOnWriteArraySet<RegistryNodeInfo> connectedProviderNodeSet = new CopyOnWriteArraySet<>();
    private final ReentrantLock lock = new ReentrantLock();
    private final Condition connected = lock.newCondition();
    private static final long waitTimeout = 3000L;
    private final RpcLoadBalance loadBalance = new RpcLoadBalanceRoundRobin();
    private volatile boolean isRunning = true;

    private ConnectionManager() {
    }

    private static class SingletonHolder {
        private static final ConnectionManager instance = new ConnectionManager();
    }

    public static ConnectionManager getInstance() {
        return SingletonHolder.instance;
    }

    /**
     * 更新连接
     * 注：RegistryNodeInfo = host + port + serviceList 地址+服务列表
     * @param serviceList zk所有子节点的路径集合
     */
    public void updateConnectedServer(List<RegistryNodeInfo> serviceList) {
        if (serviceList != null && serviceList.size() > 0) {
            Set<RegistryNodeInfo> serviceSet = new HashSet<>(serviceList);
            //如果出现新的注册信息，则连接
            for (final RegistryNodeInfo registryNodeInfo : serviceSet) {
                if (!connectedProviderNodeSet.contains(registryNodeInfo)) {
                    connectServerNode(registryNodeInfo);
                }
            }
            //如果现有注册信息不在传进来的参数中，代表这个注册信息无效，则关闭连接
            for (RegistryNodeInfo registryNodeInfo : connectedProviderNodeSet) {
                if (!serviceSet.contains(registryNodeInfo)) {
                    log.info("Remove invalid service: " + registryNodeInfo.toJson());
                    removeAndCloseHandler(registryNodeInfo);
                }
            }
        } else {
            // 传入的参数为空，代表没有有效注册信息，则关闭所有连接
            log.error("No available service!");
            for (RegistryNodeInfo registryNodeInfo : connectedProviderNodeSet) {
                removeAndCloseHandler(registryNodeInfo);
            }
        }
    }

    /**
     * 更新连接的重载，根据指定的事件（zk节点更新事件）
     * @param registryNodeInfo
     * @param type
     */
    public void updateConnectedServer(RegistryNodeInfo registryNodeInfo, PathChildrenCacheEvent.Type type) {
        if (registryNodeInfo == null) {
            return;
        }
        if (type == PathChildrenCacheEvent.Type.CHILD_ADDED && !connectedProviderNodeSet.contains(registryNodeInfo)) {
            connectServerNode(registryNodeInfo);
        } else if (type == PathChildrenCacheEvent.Type.CHILD_UPDATED) {
            removeAndCloseHandler(registryNodeInfo);
            connectServerNode(registryNodeInfo);
        } else if (type == PathChildrenCacheEvent.Type.CHILD_REMOVED) {
            removeAndCloseHandler(registryNodeInfo);
        } else {
            throw new IllegalArgumentException("Unknow type:" + type);
        }
    }

    /**
     * 根据给出的注册信息连接该节点
     * @param registryNodeInfo
     */
    private void connectServerNode(RegistryNodeInfo registryNodeInfo) {
        if (registryNodeInfo.getServiceInfoList() == null || registryNodeInfo.getServiceInfoList().isEmpty()) {
            log.info("No service on node, host: {}, port: {}", registryNodeInfo.getHost(), registryNodeInfo.getPort());
            return;
        }
        log.info("New service node, host: {}, port: {}", registryNodeInfo.getHost(), registryNodeInfo.getPort());
        /**
         * 日志打印新增注册信息的服务列表
         */
        for (RpcServiceInfo serviceInfo : registryNodeInfo.getServiceInfoList()) {
            log.info("New service info, interfaceName: {}, version: {}", serviceInfo.getInterfaceName(), serviceInfo.getServiceName());
        }
        final InetSocketAddress remotePeer = new InetSocketAddress(registryNodeInfo.getHost(), registryNodeInfo.getPort());
        //提交线程池，执行连接
        threadPoolExecutor.submit(() -> {
            Bootstrap b = new Bootstrap();
            b.group(eventLoopGroup)
                    .channel(NioSocketChannel.class)
                    .handler(new RpcClientChannelInitializer());
            ChannelFuture channelFuture = b.connect(remotePeer);
            channelFuture.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(final ChannelFuture channelFuture) throws Exception {
                    if (channelFuture.isSuccess()) {
                        /**
                         * 将连接到的地址加入集合中
                         */
                        connectedProviderNodeSet.add(registryNodeInfo);
                        log.info("Successfully connect to remote server, remote peer = " + remotePeer);
                        RpcClientHandler handler = channelFuture.channel().pipeline().get(RpcClientHandler.class);
                        connectedServerNodes.put(registryNodeInfo, handler);
                        handler.setRegistryNodeInfo(registryNodeInfo);
                        //通知等待的线程已完成一个连接
                        signalAvailableHandler();
                    } else {
                        log.error("Can not connect to remote server, remote peer = " + remotePeer);
                    }
                }
            });
        });
    }

    private void removeAndCloseHandler(RegistryNodeInfo registryNodeInfo) {
        RpcClientHandler handler = connectedServerNodes.get(registryNodeInfo);
        if (handler != null) {
            handler.close();
        }
        connectedServerNodes.remove(registryNodeInfo);
        connectedProviderNodeSet.remove(registryNodeInfo);
    }

    public void removeHandler(RegistryNodeInfo registryNodeInfo) {
        connectedProviderNodeSet.remove(registryNodeInfo);
        connectedServerNodes.remove(registryNodeInfo);
        log.info("Remove one connection, host: {}, port: {}", registryNodeInfo.getHost(), registryNodeInfo.getPort());
    }

    public void stop() {
        isRunning = false;
        for (RegistryNodeInfo registryNodeInfo : connectedProviderNodeSet) {
            removeAndCloseHandler(registryNodeInfo);
        }
        signalAvailableHandler();
        threadPoolExecutor.shutdown();
        eventLoopGroup.shutdownGracefully();
    }

    /**
     * 通知其他线程连接操作完成
     */
    private void signalAvailableHandler() {
        lock.lock();
        try {
            connected.signalAll();
        } finally {
            lock.unlock();
        }
    }

    /**
     * 让当前请求连接管理器的线程等待一段时间，看看有没有连接好的
     * @return
     * @throws InterruptedException
     */
    private boolean waitingForHandler() throws InterruptedException {
        lock.lock();
        try {
            log.warn("Waiting for available service");
            return connected.await(waitTimeout, TimeUnit.MILLISECONDS);
        } finally {
            lock.unlock();
        }
    }

    /**
     * 发送请求时执行负载均衡，在选择时需要进行线程同步
     * @param serviceKey
     * @return
     * @throws Exception
     */
    public RpcClientHandler chooseHandler(String serviceKey) throws Exception {
        int size = connectedServerNodes.values().size();
        while (isRunning && size <= 0) {
            try {
                waitingForHandler();
                size = connectedServerNodes.values().size();
            } catch (InterruptedException e) {
                log.error("Waiting for available service is interrupted!", e);
            }
        }
        RegistryNodeInfo registryNodeInfo = loadBalance.route(serviceKey, connectedServerNodes);
        RpcClientHandler handler = connectedServerNodes.get(registryNodeInfo);
        if (handler != null) {
            return handler;
        } else {
            throw new Exception("Can not get available connection");
        }
    }
}
